package drds.plus.repository.berkeley.spi;

import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.rep.ReplicaWriteException;
import drds.plus.common.lifecycle.AbstractLifecycle;
import drds.plus.common.model.DataNode;
import drds.plus.common.properties.ConnectionProperties;
import drds.plus.executor.command_handler.CommandHandlerFactory;
import drds.plus.executor.cursor.cursor_factory.CursorFactory;
import drds.plus.executor.cursor.cursor_factory.CursorFactoryImpl;
import drds.plus.executor.data_node_executor.spi.DataNodeExecutor;
import drds.plus.executor.repository.Repository;
import drds.plus.executor.table.ITable;
import drds.plus.executor.table.ITemporaryTable;
import drds.plus.sql_process.abstract_syntax_tree.configuration.TableMetaData;
import drds.plus.util.GeneralUtil;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;

import java.io.File;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class RepositoryImpl extends AbstractLifecycle implements Repository {
    public Logger log() {
        return log;
    }

    static AtomicLong currentID = new AtomicLong(0L);

    private final Map map;
    protected CommandHandlerFactory commandHandlerFactory = null;
    protected BDBConfig bdbConfig;
    protected Environment environment;
    protected Map<String, ITable> tableNameToTableMap = new ConcurrentHashMap<String, ITable>();
    protected CursorFactory cursorFactory;
    protected Environment temporaryEnvironment;
    protected Random random = new Random();

    public RepositoryImpl(Map map) {
        this.map = map;
    }

    public static long genRequestId() {
        return currentID.addAndGet(1L);
    }

    public void commonConfig(EnvironmentConfig environmentConfig) {
        System.setProperty("JEMonitor", "true");
        environmentConfig.setConfigParam(EnvironmentConfig.NODE_MAX_ENTRIES, "256");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_EVICT_BYTES, (1024 * 1024 * 2) + "");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_NODES_PER_SCAN, "10");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_LRU_ONLY, "false");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_FORCED_YIELD, "true");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_CORE_THREADS, Runtime.getRuntime().availableProcessors() + "");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_MAX_THREADS, Runtime.getRuntime().availableProcessors() + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CHECKPOINTER_BYTES_INTERVAL, 1024 * 1024 * 200 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_LAZY_MIGRATION, "true");
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_THREADS, Runtime.getRuntime().availableProcessors() + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_LOOK_AHEAD_CACHE_SIZE, 1024 * 8 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_READ_SIZE, 1024 * 1024 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_FILE_MAX, 1024 * 1024 * 200 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_FILE_CACHE_SIZE, "1024");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_USE_WRITE_QUEUE, "true");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_WRITE_QUEUE_SIZE, 1024 * 1024 * 2 + "");
        // envConfig.setConfigParam(EnvironmentConfig.HALT_ON_COMMIT_AFTER_CHECKSUMEXCEPTION,
        // "true");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_ITERATOR_READ_SIZE, 1024 * 8 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.LOCK_TIMEOUT, 2000 + "\tMILLISECONDS");
        environmentConfig.setConfigParam(EnvironmentConfig.ENV_RECOVERY_FORCE_CHECKPOINT, "true");
    }

    public void commonConfig(EnvironmentConfig environmentConfig, BDBConfig bdbConfig) {
        System.setProperty("JEMonitor", "true");
        environmentConfig.setConfigParam(EnvironmentConfig.NODE_MAX_ENTRIES, "256");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_EVICT_BYTES, (1024 * 1024 * 2) + "");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_NODES_PER_SCAN, "10");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_LRU_ONLY, "false");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_FORCED_YIELD, "true");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_CORE_THREADS, Runtime.getRuntime().availableProcessors() + "");
        environmentConfig.setConfigParam(EnvironmentConfig.EVICTOR_MAX_THREADS, Runtime.getRuntime().availableProcessors() + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CHECKPOINTER_BYTES_INTERVAL, 1024 * 1024 * 200 + "");

        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_LOOK_AHEAD_CACHE_SIZE, 1024 * 8 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_READ_SIZE, 1024 * 1024 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_MAX_BATCH_FILES, 3 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_FILE_MAX, 1024 * 1024 * 200 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_FILE_CACHE_SIZE, "1024");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_USE_WRITE_QUEUE, "true");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_WRITE_QUEUE_SIZE, 1024 * 1024 * 2 + "");
        // envConfig.setConfigParam(EnvironmentConfig.HALT_ON_COMMIT_AFTER_CHECKSUMEXCEPTION,
        // "true");
        environmentConfig.setConfigParam(EnvironmentConfig.LOG_ITERATOR_READ_SIZE, 1024 * 8 + "");
        environmentConfig.setConfigParam(EnvironmentConfig.LOCK_TIMEOUT, 2000 + "\tMILLISECONDS");
        environmentConfig.setConfigParam(EnvironmentConfig.ENV_RECOVERY_FORCE_CHECKPOINT, "true");

        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_MIN_UTILIZATION, (bdbConfig.getCleaner_min_utilization()));
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_LAZY_MIGRATION, bdbConfig.getCleanerLazyMigration());
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_THREADS, bdbConfig.getCleanerThreadCount() + "");
        environmentConfig.setConfigParam(EnvironmentConfig.CLEANER_MAX_BATCH_FILES, bdbConfig.getCleanerBatchFileCount() + "");
        environmentConfig.setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, bdbConfig.getAutoClean() ? "true" : "false");

    }

    public BDBConfig getRepositoryConfig() {
        return bdbConfig;
    }

    public CursorFactory getCursorFactory() {
        return cursorFactory;
    }

    public void init() {
        commandHandlerFactory = new CommandHandlerFactoryImpl();
        cursorFactory = new CursorFactoryImpl();
    }


    public ITable getTable(String dataNodeId, TableMetaData tableMetaData) throws RuntimeException {
        ITable table = tableNameToTableMap.get(tableMetaData.getTableName());
        if (table == null) {
            synchronized (this) {
                table = tableNameToTableMap.get(tableMetaData.getTableName());
                if (table == null) {
                    try {
                        table = initTable(tableMetaData);
                    } catch (ReplicaWriteException ex) {
                        throw new RuntimeException(ex);
                    }
                    if (!tableMetaData.isTemporaryTable()) {
                        tableNameToTableMap.put(tableMetaData.getTableName(), table);
                    }
                }

            }
        }
        return table;
    }

    public Table initTable(TableMetaData tableMetaData) throws RuntimeException {
        return new Table(tableMetaData, this);
    }

    public Database getDatabase(String tableName, boolean isTemporaryTable, boolean isSortedDuplicates) {
        DatabaseConfig databaseConfig = new DatabaseConfig();
        databaseConfig.setAllowCreate(true);
        Environment environment = this.environment;
        if (isTemporaryTable) {
            databaseConfig.setTemporary(true);
            databaseConfig.setSortedDuplicates(isSortedDuplicates);
            environment = getTemporaryEnvironment();
        } else {
            if (!bdbConfig.isTransactional()) {
                databaseConfig.setDeferredWrite(bdbConfig.isCommitSync());
            } else {
                databaseConfig.setTransactional(true);
            }
        }
        Database database = buildPrimaryIndex(tableName, databaseConfig, environment);
        return database;
    }

    public Environment getEnvironment() {
        return environment;
    }

    public void doDestroy() throws RuntimeException {
        for (Entry<String, ITable> entry : tableNameToTableMap.entrySet()) {
            entry.getValue().close();
        }
        environment.close();
        if (temporaryEnvironment != null) {
            temporaryEnvironment.close();
        }
    }

    public synchronized Environment getTemporaryEnvironment() {
        if (temporaryEnvironment != null) {
            return temporaryEnvironment;
        } else {
            EnvironmentConfig environmentConfig = new EnvironmentConfig();
            long cachePercent = GeneralUtil.getExtraCmdLong(map, ConnectionProperties.TEMP_TABLE_CACHE_PERCENT, 20);
            environmentConfig.setCachePercent((int) cachePercent);
            environmentConfig.setAllowCreate(true);
            environmentConfig.setConfigParam(EnvironmentConfig.LOG_FILE_MAX, 1000000 + "");
            String path = GeneralUtil.getExtraCmdString(map, ConnectionProperties.TEMP_TABLE_DIR);
            if (path == null) {
                path = System.getProperty("user.dir") + "/bdbtmp/" + System.currentTimeMillis() + random.nextInt() + "requestID." + genRequestId() + "/";
            }
            File file = new File(path);
            // repo_dir = new File(System.getProperty("user.dir") + "/bdbtmp/");
            if (!file.exists()) {
                file.mkdirs();
            }
            temporaryEnvironment = new Environment(file, environmentConfig);
            return temporaryEnvironment;
        }
    }


    private Database buildPrimaryIndex(String tableName, DatabaseConfig databaseConfig, Environment environment) {
        Database database = environment.openDatabase(null, tableName, databaseConfig);
        return database;
    }

    //
    public void cleanTemporaryEnvironmentLog() {
        if (temporaryEnvironment != null) {
            temporaryEnvironment.cleanLog();
        }
    }

    public int cleanEnvironmentLog() {
        return environment.cleanLog();
    }

    public CommandHandlerFactory getCommandExecutorFactory() {
        return commandHandlerFactory;
    }

    public DataNodeExecutor getDataNodeExecutor(DataNode dataNode) {
        throw new UnsupportedOperationException();
    }

    public void setBdbConfig(BDBConfig bdbConfig) {
        this.bdbConfig = bdbConfig;

    }

    public ITemporaryTable getTemporaryTable(TableMetaData tableMetaData) throws RuntimeException {
        TemporaryTable temporaryTable = new TemporaryTable(tableMetaData, this);
        temporaryTable.environment = this.getTemporaryEnvironment();
        return temporaryTable;
    }

}
